package com.example.review;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * Created with IntelliJ IDEA.
 * ClassName: CustomProducerCallBack
 * Package: com.example.review
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-08-05
 * Time: 10:44
 */

//带回调的生产者
public class CustomProducerCallBack {

    public static void main(String[] args) {

        //创建配置文件
        Properties properties = new Properties();

        //设置连接信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");

        //设置序列化器
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //1.创建生产者
        KafkaProducer<String, String> kafkaProducer =
                new KafkaProducer<String, String>(properties);

        //2.发送数据
        kafkaProducer.send(new ProducerRecord<>("first", "Java"), new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                //recordMetadata 是数据信息  Exception异常信息
                if(e == null){
                    //如果e为空说明发送正常
                    //可以获取数据信息 分区 或者 主题
                    System.out.println(recordMetadata.topic());
                }
            }
        });

        //关闭资源
        kafkaProducer.close();

    }

}
